package com.rpc.handler;

import com.google.protobuf.ByteString;
import com.netty.client.socket.ImSocketClient;
import com.netty.client.socket.ImSocketClientHandler;
import com.netty.constant.Constants;
import com.netty.model.proto.MessageBodyProto;
import com.netty.utils.Md5Utils;
import com.rpc.common.Invocation;
import com.rpc.common.UnprocessedRequests;
import com.rpc.utils.ProtostuffUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

public class RpcClientHandler implements ImSocketClientHandler {
    private Logger logger = LoggerFactory.getLogger(getClass());

    public Map<String,Channel> maps = new HashMap<>();
    private UnprocessedRequests unprocessedRequests = new UnprocessedRequests();
    private String clientKey = "client-"+ Md5Utils.shortMd5(UUID.randomUUID().toString().replaceAll("-",""));
    private ImSocketClient imSocketClient = null;

    public void startClient(String url, int port){
        imSocketClient = new ImSocketClient(url,port,this);
        imSocketClient.startClient();
    }
    public void startClient(String url, int port,Map<String,Channel> maps){
        this.maps = maps;
        imSocketClient = new ImSocketClient(url,port,this);
        imSocketClient.startClient();
    }


    @Override
    public void channelActive(ChannelHandlerContext channelHandlerContext) {
        MessageBodyProto.Model.Builder body = MessageBodyProto.Model.newBuilder();
        body.setType(Constants.CmdType.ONLINE);
        body.setTimestamp(new Date().getTime());
        body.setKey(clientKey);
        body.putData("channel", ByteString.copyFrom(Constants.ImserverConfig.SOCKET.getBytes()));
        channelHandlerContext.writeAndFlush(body);
    }

    @Override
    public void channelRead0(Channel channel, MessageBodyProto.Model replyBody) {
        String channelId = channel.id().toString();
        maps.put(clientKey+channelId,channel);
        if(replyBody.getDataMap().keySet().size() > 0){
            String requestId = replyBody.getDataMap().get("requestId").toStringUtf8();
            String time = replyBody.getDataMap().get("time").toStringUtf8();
            long times = System.currentTimeMillis() - Long.parseLong(time);
            logger.info("获取请求id【{}】,共耗时为:【{}】",requestId,times);
            unprocessedRequests.complete(requestId,ProtostuffUtils.deserialize(replyBody.getDataMap().get("data").toByteArray()));
        }
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext channelHandlerContext) {
        String channelId = channelHandlerContext.channel().id().toString();
        maps.remove(clientKey+channelId);
    }

    public CompletableFuture sendRpcService(String serverName, Invocation invocation){
        String uuid= UUID.randomUUID().toString().replaceAll("-","");
        List<String> keys =maps.keySet().stream().collect(Collectors.toList());
        String uid = keys.get(new Random().nextInt(keys.size()));
        Channel channel = maps.get(uid);
        logger.info("获取请求id【{}】",uuid);
        CompletableFuture<Object> future  = new CompletableFuture<>();
        unprocessedRequests.put(uuid,future);
        MessageBodyProto.Model.Builder builder = MessageBodyProto.Model.newBuilder();
        builder.setType(Constants.CmdType.MESSAGE);
        builder.setKey(clientKey);
        builder.putData("serverName",ByteString.copyFrom(serverName.getBytes(StandardCharsets.UTF_8)));
        builder.putData("requestId",ByteString.copyFrom(uuid.getBytes(StandardCharsets.UTF_8)));
        builder.putData("time",ByteString.copyFrom((System.currentTimeMillis()+"").getBytes(StandardCharsets.UTF_8)));
        builder.putData("invocation", ByteString.copyFrom(ProtostuffUtils.serialize(invocation)));
        channel.writeAndFlush(builder);
        return future;
    }

}
